package com.zhiyu.middle.rocketmq.service.impl;

import com.zhiyu.middle.rocketmq.entity.MessageReq;
import com.zhiyu.middle.rocketmq.service.RocketMQService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;

@Slf4j
@Service
public class RocketMQServiceImpl implements RocketMQService {

    @Resource
    private RocketMQTemplate rocketMQTemplate;


    /**
     * RocketMQ消息同步发送
     * 指消息发送方发出数据后，会在收到接收方发回响应之后才发下一个数据包的通讯方式。
     */
    @Override
    public void syncRocketMQMessageSend(MessageReq messageReq) {
        Message message = MessageBuilder.withPayload(messageReq.getMessage()).build();
        SendResult sendResultObj = rocketMQTemplate.syncSend(messageReq.getTopic(), message);
        log.info("MQ同步发送对象类型的消息topic为:{},返回结果:{}", messageReq.getTopic(), sendResultObj);
    }

    /**
     * 异步发送:
     * 指发送方发出数据后，不等接收方发回响应，接着发送下个数据包的通讯方式。
     * MQ 的异步发送，需要用户实现异步发送回调接口（SendCallback），在执行消息的异步发送时，
     * 应用不需要等待服务器响应即可直接返回，通过回调接口接收务器响应，并对服务器的响应结果进行处理。
     */
    @Override
    public void asyncRocketMQMessageSend(MessageReq messageReq) {
        Message message = MessageBuilder.withPayload(messageReq.getMessage()).build();
        rocketMQTemplate.asyncSend(messageReq.getTopic(), message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("异步消息发送成功:{}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("异步消息发送失败:{}", throwable.getCause());
            }
        });
    }

    /**
     * 特点为只负责发送消息，不等待服务器回应且没有回调函数触发，即只发送请求不等待应答。
     * 此方式发送消息的过程耗时非常短，一般在微秒级别。
     * 应用场景：适用于某些耗时非常短，但对可靠性要求并不高的场景，例如日志收集。
     * 与异步发送的区别在于:异步其实还是要发送结果，不过是回调回来，不阻塞当前线程等待结果。
     */
    @Override
    public void oneWayRocketMQMessageSend(MessageReq messageReq) {
        Message message = MessageBuilder.withPayload(messageReq.getMessage()).build();
        rocketMQTemplate.sendOneWay(messageReq.getTopic(), message);
    }

    /**
     * 延迟消息：
     * rocketMQ的延迟消息发送其实是一发送就已经到broker端了，然后消费端会延迟收到消息。
     * RocketMQ 目前只支持固定精度的定时消息。
     * 延迟级别（18个等级）
     * 1到18分别对应1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
     * 为什么不支持固定精度的定时消息？
     * 因为rocketMQ之所以性能好的原因是broker端做的事情很少，基本都交给业务端做，就是消费端。如果支持
     * 我们自定义的延迟时候，会很大程度消耗broker的性能。
     * <p>
     * 延迟的底层方法是用定时任务实现的。
     */
    @Override
    public void delayedRocketMQMessageSend(MessageReq messageReq) {
        Message message = MessageBuilder.withPayload(messageReq.getMessage()).build();
        SendResult sendResult = rocketMQTemplate.syncSend(messageReq.getTopic(), message, 1000, 16);
        log.info("发送延迟消息返回结果:{}", sendResult);
    }

    @Override
    public void orderlyRocketMQMessageSend(MessageReq messageReq, String key) {
        Message message = MessageBuilder.withPayload(messageReq.getMessage()).build();
        // 同步顺序发送 - 对应顺序消费：
        SendResult sendResult = rocketMQTemplate.syncSendOrderly(messageReq.getTopic(), message, key);
        log.info("同步发送顺序消息返回结果:{}", sendResult);

        // 异步顺序发送 - 对应顺序消费：
        rocketMQTemplate.asyncSendOrderly(messageReq.getTopic(), message, key, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("异步发送顺序返回结果为:{}", sendResult);
            }

            @Override
            public void onException(Throwable e) {
                log.error("异步发送顺序消息失败，原因为:{}", e.getCause());
            }
        });
    }

    @Override
    public void selectorRocketMQMessageSend(MessageReq messageReq) {
        Message message = MessageBuilder.withPayload(messageReq.getMessage()).build();
        SendResult sendResult = rocketMQTemplate.syncSend(messageReq.getTopic() + messageReq.getTag(), message);
        log.info("消息过滤发送成功返回结果为:{}，当前k的参数值为:{}", sendResult, 0);
    }

    /**
     * 分布式事务消息发送
     */
    @Override
    public void transactionMQSend(MessageReq messageReq, String key) {
        // 消息发送回调：TransactionRocketMQListener
        String txProducerGroup = "transaction-group";
        // topic:tag
        String destination = "tx-text-topic";
        Message message = MessageBuilder
                .withPayload(messageReq.getMessage())
                .setHeader(RocketMQHeaders.TRANSACTION_ID, UUID.randomUUID().toString())
                .setHeader(RocketMQHeaders.KEYS, key)
                .build();
        // 同步阻塞
        CountDownLatch latch = new CountDownLatch(1);
        TransactionSendResult sendResult = rocketMQTemplate.sendMessageInTransaction(txProducerGroup, destination, message, latch);
        if (sendResult.getSendStatus().equals(SendStatus.SEND_OK) && sendResult.getLocalTransactionState().equals(LocalTransactionState.COMMIT_MESSAGE)) {
            // 下单成功，并且消息对消费端可见。

            // 在这里可以异步通知上游服务，也可以继续走自己的逻辑，比如有些逻辑必须保证下单和库存成功之后才能走的。

            log.info("消息发送成功，并且本地事务执行成功");
        }
        try {
            latch.await();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}
